package com.swak.reactivex.p2p.exchange;

import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.reactivex.p2p.Group;
import com.swak.reactivex.p2p.Networker;
import com.swak.reactivex.p2p.Networker.NetworkerListener;
import com.swak.reactivex.p2p.PeerID;
import com.swak.reactivex.transport.ChannelHandler;
import com.swak.reactivex.transport.ExchangeClient;
import com.swak.reactivex.transport.ExchangeHandler;
import com.swak.reactivex.transport.ExchangeHandlerDispatcher;
import com.swak.reactivex.transport.ExchangeServer;
import com.swak.reactivex.transport.Exchangers;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.spi.ExtensionLoader;

/**
 * ExchangeGroup.
 * 
 * @author 618lf
 */
@Deprecated
public class ExchangeGroup implements Group {

	private static final Logger logger = LoggerFactory.getLogger(ExchangeGroup.class);
	private static ExtensionLoader<Networker> NETWORKER = ExtensionLoader.getExtensionLoader(Networker.class);

	private final URL url;
	private final PeerID peerID;
	private final LoopResources resources;
	private final Networker networker;
	private final Map<URL, ExchangeServer> servers = new ConcurrentHashMap<URL, ExchangeServer>();
	private final Map<URL, ExchangeClient> clients = new ConcurrentHashMap<URL, ExchangeClient>();
	private final ExchangeChannels channels = new ExchangeChannels();
	private final ExchangeHandlerDispatcher dispatcher;

	public ExchangeGroup(URL url, LoopResources resources) {
		this.url = url;
		this.peerID = PeerID.localPeerID();
		this.networker = NETWORKER.load(url.getParameter(Constants.Networker_Key), new Object[] { url },
				this.getClass().getClassLoader());
		this.resources = resources;
		this.networker.listen(new NetworkerListener() {

			@Override
			public void join(URL url) throws RemotingException {
				connect(url);
			}

			@Override
			public void leave(URL url) throws RemotingException {
				disconnect(url);
			}
		});
		this.dispatcher = new ExchangeHandlerDispatcher();
	}

	@Override
	public URL getUrl() {
		return url;
	}

	@Override
	public ExchangePeer join(URL url, ChannelHandler handler) throws RemotingException {
		return join(url, (ExchangeHandler) handler);
	}

	protected ExchangePeer join(URL url, ExchangeHandler handler) throws RemotingException {
		URL peerUrl = URL.valueOf(url).addParameter(Constants.Peer_Id, this.peerID.getId());
		PeerID peerID = new PeerID().setId(this.peerID.getId()).setUrl(peerUrl.toFullString());
		ExchangeGroupHandler groupHandler = new ExchangeGroupHandler(peerID, this, handler, clients, channels);
		ExchangeServer server = servers.get(peerUrl);
		if (server == null) {
			server = Exchangers.bind(peerUrl, resources, groupHandler);
			servers.put(peerUrl, server);
			this.dispatcher.addChannelHandler(groupHandler);
		}
		this.networker.join(peerUrl);
		return new ExchangePeer(peerID, this, server, clients, channels);
	}

	@Override
	public void leave(URL url) throws RemotingException {
		ExchangeServer server = servers.get(url);
		if (server != null) {
			server.close();
		}
		this.networker.leave(url);
	}

	@Override
	public void close() {
		for (URL url : new ArrayList<URL>(servers.keySet())) {
			try {
				leave(url);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
		for (URL url : new ArrayList<URL>(clients.keySet())) {
			try {
				disconnect(url);
			} catch (Throwable t) {
				logger.error(t.getMessage(), t);
			}
		}
		try {
			this.networker.close();
		} catch (Exception t) {
			logger.error(t.getMessage(), t);
		}
	}

	protected ExchangeClient connect(URL url) throws RemotingException {
		if (servers.containsKey(url)) {
			return null;
		}
		PeerID peerID = PeerID.remotePeerID(url);
		if (channels.containsKey(peerID)) {
			return null;
		}
		ExchangeClient client = clients.get(url);
		if (client == null) {
			client = Exchangers.connect(url, resources, this.dispatcher);
			clients.put(url, client);
		}
		return client;
	}

	protected void disconnect(URL url) throws RemotingException {
		ExchangeClient client = clients.remove(url);
		if (client != null) {
			client.close();
		}
	}
}
